package com.gzhryc.common.kafka.procuder;


import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import com.gzhryc.common.kafka.JFKafkaProducer;
import com.gzhryc.common.kafka.models.KafkaMessage;

public class SingleKafkaProducer implements JFKafkaProducer {
	
	private KafkaProducer<String, String> producer = null;
	
	public SingleKafkaProducer(Properties properties) {
		producer = new KafkaProducer<String,String>(properties);
	}

	@Override
	public boolean send(KafkaMessage msg) {
		if(producer != null) {
			try {
				ProducerRecord<String, String> producerRecord = new ProducerRecord<>(msg.getTopic(), msg.getKey(), msg.getValue()) ;
				producer.send(producerRecord);
				producer.flush();
				return true;
			}catch(Exception e) {
				e.printStackTrace();
			}
		}
		return false;
	}

	@Override
	public void close() {
		producer.close();
	}

	
}
